package com.atguigu.tingshu.cdc.consumer;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.atguigu.tingshu.cdc.entity.CDCEntity;
import com.atguigu.tingshu.cdc.handler.EntryHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;
import java.util.List;

/**
 * CDC 消费者类，负责连接 Canal 服务器并处理 binlog 事件
 */
@Component
@Slf4j
public class CdcConsumer {

    // 注入 RedisTemplate 用于操作 Redis 缓存
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    // 注入 RabbitTemplate 用于发送消息到 RabbitMQ
    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 注入 EntryHandler 用于处理 CDC 事件
    @Autowired
    private EntryHandler cdcEntryHandler;

    @Autowired
    private CanalConnector connector;

    /**
     * 监听 album_info 表中 id 字段的变化
     */
    @PostConstruct
    public void listenAlbumInfo() {
        while (true) { // 无限循环，持续监听 binlog 事件
            try {
                // 从 Canal 服务器获取最多 100 条未确认的消息
                Message message = connector.getWithoutAck(10);
                // 获取当前批次 ID
                long batchId = message.getId();
                // 获取当前批次中的条目数量
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) { // 如果没有新消息
                    try {
                        // 线程休眠 1 秒钟
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        // 记录线程睡眠中断的错误日志
                        log.error("线程睡眠时发生中断", e);
                        break; // 中断后退出循环
                    }
                } else {
                    // 处理获取到的 binlog 事件
                    printEntry(message.getEntries());
                }
                // 提交确认，告知 Canal 服务器已处理完当前批次
                connector.ack(batchId);
            } catch (Exception e) {
                // 记录监听 binlog 事件时发生的错误日志
                log.error("监听 binlog 事件时发生错误", e);
                // 断开连接并重新连接 Canal 服务器
                connector.disconnect();
                connector.connect();
                connector.rollback();
            }
        }
    }

    /**
     * 打印并处理 binlog 事件中的每个条目
     *
     * @param entrys binlog 事件条目列表
     */
    private void printEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            // 跳过事务开始和结束的条目
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChage = null;
            try {
                // 解析 binlog 事件条目的存储值为 RowChange 对象
                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                // 如果解析失败，抛出运行时异常并记录错误信息
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
            }

            // 获取当前条目的事件类型（插入、更新、删除等）
            CanalEntry.EventType eventType = rowChage.getEventType();
            // 记录 binlog 事件的详细信息
            log.info("================> binlog[{}:{}] , name[{},{}] , eventType : {}",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType);

            // 遍历当前条目中的所有行数据
            for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                // 根据事件类型调用相应的处理方法
                if (eventType == CanalEntry.EventType.DELETE) {
                    handleDelete(rowData.getBeforeColumnsList());
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    handleInsert(rowData.getAfterColumnsList());
                } else if (eventType == CanalEntry.EventType.UPDATE) {
                    handleUpdate(rowData.getBeforeColumnsList(), rowData.getAfterColumnsList());
                }
            }
        }
    }

    /**
     * 处理插入事件
     *
     * @param columns 插入事件中的列数据
     */
    private void handleInsert(List<CanalEntry.Column> columns) {
        CDCEntity cdcEntity = new CDCEntity(); // 创建新的 CDCEntity 实例
        for (CanalEntry.Column column : columns) {
            // 如果列名为 "id"，设置实体的 ID 属性
            if ("id".equals(column.getName())) {
                cdcEntity.setId(Long.parseLong(column.getValue()));
            }
            // 设置其他字段...
        }
        // 调用 EntryHandler 的 insert 方法处理插入事件
        cdcEntryHandler.insert(cdcEntity);
    }

    /**
     * 处理更新事件
     *
     * @param beforeColumns 更新前的列数据
     * @param afterColumns  更新后的列数据
     */
    private void handleUpdate(List<CanalEntry.Column> beforeColumns, List<CanalEntry.Column> afterColumns) {
        CDCEntity beforeEntity = new CDCEntity(); // 创建更新前的 CDCEntity 实例
        CDCEntity afterEntity = new CDCEntity(); // 创建更新后的 CDCEntity 实例
        for (CanalEntry.Column column : beforeColumns) {
            // 如果列名为 "id"，设置更新前实体的 ID 属性
            if ("id".equals(column.getName())) {
                beforeEntity.setId(Long.parseLong(column.getValue()));
            }
            // 设置其他字段...
        }
        for (CanalEntry.Column column : afterColumns) {
            // 如果列名为 "id"，设置更新后实体的 ID 属性
            if ("id".equals(column.getName())) {
                afterEntity.setId(Long.parseLong(column.getValue()));
            }
            // 设置其他字段...
        }
        // 调用 EntryHandler 的 update 方法处理更新事件
        cdcEntryHandler.update(beforeEntity, afterEntity);
    }

    /**
     * 处理删除事件
     *
     * @param columns 删除事件中的列数据
     */
    private void handleDelete(List<CanalEntry.Column> columns) {
        CDCEntity cdcEntity = new CDCEntity(); // 创建新的 CDCEntity 实例
        for (CanalEntry.Column column : columns) {
            // 如果列名为 "id"，设置实体的 ID 属性
            if ("id".equals(column.getName())) {
                cdcEntity.setId(Long.parseLong(column.getValue()));
            }
            // 设置其他字段...
        }
        // 调用 EntryHandler 的 delete 方法处理删除事件
        Boolean delete = cdcEntryHandler.delete(cdcEntity);
        if (delete) {
            // 如果删除成功，记录成功日志
            log.info("删除成功, ID: {}", cdcEntity.getId());
        } else {
            // 如果删除失败，记录警告日志并重新发送消息到 RabbitMQ 进行重试
            log.warn("删除失败, ID: {}", cdcEntity.getId());
            rabbitTemplate.convertAndSend("retryQueue", cdcEntity.getId());
        }
    }

    /**
     * 消费消息（继续删除 Redis）
     *
     * @param cacheKey Redis 缓存键
     */
    @RabbitListener(queues = "retryQueue")
    public void consumeMessage(String cacheKey) {
        try {
            // 尝试删除 Redis 缓存
            Boolean isDeleted = redisTemplate.delete(cacheKey);
            if (Boolean.TRUE.equals(isDeleted)) {
                // 如果删除成功，记录成功日志
                log.info("成功删除缓存, key: {}", cacheKey);
            } else {
                // 如果删除失败，记录警告日志并重新发送消息到 RabbitMQ 进行重试
                log.warn("删除缓存失败, key: {}", cacheKey);
                rabbitTemplate.convertAndSend("retryQueue", cacheKey);
            }
        } catch (Exception e) {
            // 如果消费消息时发生错误，记录错误日志并重新发送消息到 RabbitMQ 进行重试
            log.error("消费消息时发生错误, key: {}", cacheKey, e);
            rabbitTemplate.convertAndSend("retryQueue", cacheKey);
        }
    }
}
